package com.atguigu.gmall.realtime.util;

import com.atguigu.gmall.realtime.common.Constant;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @Author lzc
 * @Date 2022/12/2 13:52
 */
public class FlinkSourceUtil {
    
    public static KafkaSource<String> getKafkaSource(String groupId, String topic) {
        return KafkaSource.<String>builder()
            .setBootstrapServers(Constant.KAFKA_BROKERS)
            .setTopics(topic)
            .setGroupId(groupId)
            .setStartingOffsets(OffsetsInitializer.latest())
            .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                // 反序列化. 返回值就会自动放入到流中
                @Override
                public String deserialize(byte[] message) throws IOException {
                    //  left join 有删除数据, 写入的 null, 这个时候message是 null
                    if (message != null) {
                        return new String(message, StandardCharsets.UTF_8);
                    }
                    return null;  // 如果这里返回的是 null, 这 null 不会放到流中
                }
    
                // 是否结束流
                @Override
                public boolean isEndOfStream(String nextElement) {
                    return false;  // 返回 false, 流用于不结束
                }
                
                // 返回流中的数据类型
                @Override
                public TypeInformation<String> getProducedType() {
//                    return Types.STRING;  // 特殊的类: 内置, 比如: 基本类型,string, tuple 等
//                    return TypeInformation.of(String.class);  // 普通的类: 类不能带泛型
                    return TypeInformation.of(new TypeHint<String>() {});  // 普通的类: 类不能带泛型
                }
            }) // 反序列化器
            // 为了避免消费事务关闭前的数据: 只读取已提交的数据
            .setProperty("isolation.level", "read_committed")
            .build();
        
        
    
    
    }
}
